package com.antgroup.seckill.business.common.rocketmq;

import com.alibaba.fastjson.JSON;
import com.antgroup.seckill.business.common.constants.CommonConstant;
import com.antgroup.seckill.business.common.vo.response.GoodsVo;
import com.antgroup.seckill.business.goods.model.SeckillMessage;
import com.antgroup.seckill.business.goods.service.IGoodsService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.Objects;

/**
 * @author wangxy
 * @date 2024/8/18 19:18
 */
@Service
@Slf4j
@RocketMQMessageListener(topic = "goods", consumerGroup = "goods-consumer-group")
public class GoodsRcketMqConsumer implements RocketMQListener<MessageExt> {
    @Autowired
    private IGoodsService iGoodsService;

    @Autowired
    private RestHighLevelClient esClient;

    @Override
    public void onMessage(MessageExt messageExt) {
        try {
            log.info("处理新增消息: {}", new String(messageExt.getBody()));
            String topic = messageExt.getTopic();
            String tag = messageExt.getTags();
            String message = new String(messageExt.getBody());
            SeckillMessage seckillMessage = JSON.parseObject(message, SeckillMessage.class);


            if (StringUtils.equals(tag, "CREATE")) {
                //查询商品信息 存入es
                GoodsVo goodsVo = iGoodsService.queryGoodsByGoodsId(seckillMessage.getGoogsId());

                if (Objects.isNull(goodsVo)) {
                    return;
                }

                IndexRequest indexRequest = new IndexRequest(CommonConstant.GOODS_INDEX)
                        .id(goodsVo.getId())
                        .source(JSON.toJSONString(goodsVo), XContentType.JSON);

                esClient.index(indexRequest, RequestOptions.DEFAULT);
            }

            if (StringUtils.equals(tag, "UPDATE")) {
                GoodsVo goodsVo = iGoodsService.queryGoodsByGoodsId(seckillMessage.getGoogsId());

                if (Objects.isNull(goodsVo)) {
                    return;
                }
                //更新
                UpdateRequest updateRequest = new UpdateRequest(CommonConstant.GOODS_INDEX,goodsVo.getId());
                updateRequest.doc(JSON.toJSONString(goodsVo), XContentType.JSON);
                esClient.update(updateRequest, RequestOptions.DEFAULT);
            }

            if(StringUtils.equals(tag, "DELETE")) {
                GoodsVo goodsVo = iGoodsService.queryGoodsByGoodsId(seckillMessage.getGoogsId());

                if (Objects.nonNull(goodsVo)) {
                    return;
                }

                DeleteRequest deleteRequest = new DeleteRequest(CommonConstant.GOODS_INDEX, seckillMessage.getGoogsId());
                esClient.delete(deleteRequest, RequestOptions.DEFAULT);

            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}
